package com.atguigu.tingshu.account.receiver;

import com.atguigu.tingshu.account.service.RechargeInfoService;
import com.atguigu.tingshu.account.service.UserAccountService;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.model.account.UserAccount;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.protocol.types.Field;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

/**
 * ClassName: AccoutReceiver
 * Package: com.atguigu.tingshu.account.receiver
 * Description:
 *
 * @Author:
 * @Create: 2025/7/28 - 14:28
 * @Version: v1.0
 */
@Component
@Slf4j
public class AccountReceiver {


    @Resource
    private UserAccountService userAccountService;

    @Resource
    private RechargeInfoService rechargeInfoService;

    /**
     * 监听账户初始化的消息
     */
    @KafkaListener(topics = KafkaConstant.QUEUE_USER_REGISTER)
    public void userRegister(ConsumerRecord<String,String> record){
        //得到发送的消息
        String userId = record.value();
        if(StringUtils.hasText(userId)){
            userAccountService.initUserAccount(userId);
        }
    }

    /**
     * 扣除金额
     * @param record
     */
    @KafkaListener(topics = KafkaConstant.QUEUE_ACCOUNT_MINUS)
    public void minus(ConsumerRecord<String, String> record){
        String orderNo = record.value();
        if(!StringUtils.hasText(orderNo)){
            return;
        }
        userAccountService.minus(orderNo);
    }

    /**
     * 解除锁定金额
     * @param record
     */
    @KafkaListener(topics = KafkaConstant.QUEUE_ACCOUNT_UNLOCK)
    public void unlock(ConsumerRecord<String, String> record) {
        String orderNo = record.value();
        if (StringUtils.isEmpty(orderNo)) {
            return;
        }
        //  调用解除锁定
        userAccountService.unlock(orderNo);
    }

    /**
     * 充值
     * @param record
     */
    @KafkaListener(topics = KafkaConstant.QUEUE_RECHARGE_PAY_SUCCESS)
    public void rechargePaySuccess(ConsumerRecord<String, String> record) {
        String orderNo = record.value();
        log.info("充值成功通知: {}", orderNo);
        //通知更新用户账号
        rechargeInfoService.rechargePaySuccess(orderNo);
    }
}
